package com.supermarket.spark.batch

import org.apache.spark.sql._
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types._
import java.util.Properties

object OfflineAnalysisJob {
  def main(args: Array[String]): Unit = {
    System.setProperty("hadoop.home.dir", "D:\\tmp\\hadoop")
    val spark = SparkSession.builder
      .appName("Flume Data Offline Analysis with Spark SQL")
      .master("local[*]")
      .getOrCreate()

    // 定义 schema
    val schema = StructType(Seq(
      StructField("user_id", IntegerType),
      StructField("product_id", IntegerType),
      StructField("product_name", StringType),
      StructField("in_stock", IntegerType),
      StructField("favor_level", IntegerType),
      StructField("sold", IntegerType),
      StructField("status", StringType)
    ))

    // 从 HDFS 读取 Flume 写入的原始数据
    val flumeDF = spark.read
      .option("sep", "\t")
      .schema(schema)
      .csv("hdfs://niit01:8020/user/flume/supermarket00/*")

    // 注册为临时视图
    flumeDF.createOrReplaceTempView("supermarket_raw")

    // 1. 销量 TopN 商品
    val topProductsDF = spark.sql("""
      SELECT product_id, SUM(sold) AS total_sold
      FROM supermarket_raw
      GROUP BY product_id
      ORDER BY total_sold DESC
      LIMIT 10
    """)

    // 2. 库存预警商品
    val lowStockDF = spark.sql("""
      SELECT product_id, product_name, in_stock, sold
      FROM supermarket_raw
      WHERE in_stock < 50
    """)

    // 3. 商品类别销量分布
    val categorySalesDF = spark.sql("""
      SELECT
        CASE
          WHEN product_name RLIKE 'Milk|Cheese|Butter' THEN '乳制品'
          WHEN product_name RLIKE 'Water|Juice|Cola' THEN '饮料类'
          WHEN product_name RLIKE 'Chips|Chocolate' THEN '零食类'
          WHEN product_name RLIKE 'Apple|Banana' THEN '水果类'
          WHEN product_name RLIKE 'Carrot|Tomato' THEN '蔬菜类'
          WHEN product_name RLIKE 'Detergent|Disinfectant' THEN '清洁用品'
          WHEN product_name RLIKE 'Shampoo|Toothpaste' THEN '个人护理'
          ELSE '其他'
        END AS category,
        SUM(sold) AS total_sold
      FROM supermarket_raw
      GROUP BY
        CASE
          WHEN product_name RLIKE 'Milk|Cheese|Butter' THEN '乳制品'
          WHEN product_name RLIKE 'Water|Juice|Cola' THEN '饮料类'
          WHEN product_name RLIKE 'Chips|Chocolate' THEN '零食类'
          WHEN product_name RLIKE 'Apple|Banana' THEN '水果类'
          WHEN product_name RLIKE 'Carrot|Tomato' THEN '蔬菜类'
          WHEN product_name RLIKE 'Detergent|Disinfectant' THEN '清洁用品'
          WHEN product_name RLIKE 'Shampoo|Toothpaste' THEN '个人护理'
          ELSE '其他'
        END
      ORDER BY total_sold DESC
    """)

    // MySQL 连接信息
    val jdbcUrl = "jdbc:mysql://43.140.205.103:3306/supermarket"
    val connectionProperties = new Properties()
    connectionProperties.put("user", "supermarket")
    connectionProperties.put("password", "a7NrdbX8hiAZ8Nxb")
    connectionProperties.put("driver", "com.mysql.cj.jdbc.Driver")

    // 写入 MySQL
    topProductsDF.write.mode("overwrite").jdbc(jdbcUrl, "daily_top_products", connectionProperties)
    lowStockDF.write.mode("overwrite").jdbc(jdbcUrl, "low_stock_alerts", connectionProperties)
    categorySalesDF.write.mode("overwrite").jdbc(jdbcUrl, "category_sales_distribution", connectionProperties)

    spark.stop()
  }
}